package cn.baiyang.rocketmq.spring.autoconfigure;

import cn.baiyang.rocketmq.spring.annotation.RocketMQMessageListener;
import cn.baiyang.rocketmq.spring.core.RocketMQListener;
import cn.baiyang.rocketmq.spring.support.DefaultRocketMQListenerContainer;

import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.util.StringUtils;

/**
 *
 * @author hongzhu
 * @version V1.0
 * @since 2023-07-24 18:13
 */
@Configuration
public class ListenerContainerConfiguration implements ApplicationContextAware {
    private final static Logger LOGGER = LoggerFactory.getLogger(ListenerContainerConfiguration.class);

    private ConfigurableApplicationContext applicationContext;

    private final AtomicLong counter = new AtomicLong(0);

    private ConfigurableEnvironment environment;

    private RocketMQProperties rocketMQProperties;

    public ListenerContainerConfiguration(ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties) {
        this.environment = environment;
        this.rocketMQProperties = rocketMQProperties;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

    public void registerContainer(String beanName, Object bean, RocketMQMessageListener annotation) {
        try {
            String topic = environment.resolveRequiredPlaceholders(annotation.topic());
            String consumerGroup = environment.resolveRequiredPlaceholders(annotation.consumerGroup());
            if (StringUtils.isEmpty(topic) || StringUtils.isEmpty(consumerGroup)) {
                LOGGER.info("Skip register listener cause topic or consumerGroup is empty, listenerBeanName:{}", beanName);
                return;
            }
        }catch (Exception e) {
            LOGGER.error("Skip register listener cause: {}, listenerBeanName:{}", e.getMessage(), beanName);
            return;
        }

        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);

        if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
        }

        String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
            counter.incrementAndGet());
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;

        genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
            () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
        DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
            DefaultRocketMQListenerContainer.class);
        if (!container.isRunning()) {
            try {
                container.start();
            } catch (Exception e) {
                LOGGER.error("Started container failed. {}", container, e);
                throw new RuntimeException(e);
            }
        }

        LOGGER.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
    }

    private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String containerBeanName, Object bean,
        RocketMQMessageListener annotation) {
        DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();
        container.setRocketMQMessageListener(annotation);
        container.setName(containerBeanName);

        String endpoints = environment.resolvePlaceholders(annotation.endpoints());
        endpoints = StringUtils.hasLength(endpoints) ? endpoints : rocketMQProperties.getEndpoints();
        container.setEndpoints(endpoints);
        String accesskey = environment.resolvePlaceholders(annotation.accessKey());
        accesskey = StringUtils.hasLength(accesskey) ? accesskey : rocketMQProperties.getAccesskey();
        container.setAccessKey(accesskey);
        String secretkey = environment.resolvePlaceholders(annotation.secretKey());
        secretkey = StringUtils.hasLength(secretkey) ? secretkey : rocketMQProperties.getSecretkey();
        container.setSecretKey(secretkey);
        String enableSsl = environment.resolvePlaceholders(annotation.enableSsl());
        enableSsl = StringUtils.hasLength(enableSsl) ? enableSsl :
            String.valueOf(rocketMQProperties.getEnableSsl());
        container.setEnableSsl(enableSsl);
        String requestTimeout = environment.resolvePlaceholders(annotation.requestTimeout());
        requestTimeout = StringUtils.hasLength(requestTimeout) ? requestTimeout :
            String.valueOf(rocketMQProperties.getRequestTimeout());
        container.setRequestTimeout(requestTimeout);

        container.setTopic(environment.resolvePlaceholders(annotation.topic()));
        container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));
        String selectorExpression = environment.resolvePlaceholders(annotation.selectorExpression());
        if (StringUtils.hasLength(selectorExpression)) {
            container.setSelectorExpression(selectorExpression);
        }

        if (RocketMQListener.class.isAssignableFrom(bean.getClass())) {
            container.setRocketMQListener((RocketMQListener) bean);
        }
        return container;
    }

}